package org.swift.mashup.engine.concurrency.agenda;

import java.util.Collection;
import java.util.HashSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.swift.mashup.engine.concurrency.agenda.AgendaEvent.EventType;
import org.swift.mashup.engine.concurrency.agenda.ThreadMonitor.Status;

// FIXME: these really don't belong here, get rid of them
// once the Action/Activity API is cleaned up
import org.swift.mashup.engine.concurrency.agenda.ActionException;
import org.swift.mashup.engine.model.ActivityResultItem;
import org.swift.mashup.engine.model.ActionOutput;

/**
 * @author dhesson
 */
public class ConcurrentAgenda implements Agenda {

	private static Log log = LogFactory.getLog(ConcurrentAgenda.class);

	// our helper objects
	private final Set<AgendaListener> listeners = new HashSet<AgendaListener>();
	private final TaskManager taskManager = TaskManager.getInstance();
	private final ThreadMonitor threadMonitor;

	// our execution state
	private final Set<AgendaItem> todoItems = new HashSet<AgendaItem>();
	private final Semaphore governor;
    private final int rateLimit;
	private final Map<AgendaItem, Future<?>> queuedItems;
	private int completedItems;
	private AtomicInteger startedItems;
	private AtomicInteger inProgressItems;
	private volatile boolean cancelled;
	private volatile CountDownLatch done;
	private volatile ScheduledFuture<?> scheduledTimeout;

	// our configuration
	private int agendaTimeout = 120000;
	private boolean laxErrors;

    public ConcurrentAgenda(final Collection<? extends AgendaItem> items, final AgendaConfig config) {
        if ( log.isDebugEnabled() ) {
            log.debug(config.toString());
        }

		validateArguments(items, config.getThreadMonitor(), config.getAgendaTimeout());

		this.agendaTimeout = config.getAgendaTimeout();
		this.threadMonitor = config.getThreadMonitor();
        this.rateLimit = config.getRateLimit();

		for (AgendaItem item : items) {
			if(!todoItems.add(item)) {
				log.warn("Discarding duplicate agenda item " + item);
			}
		}

		// resizing is slow, let's not do it
		final int noResizeCapacity = 2 * todoItems.size();
		queuedItems = new HashMap<AgendaItem, Future<?>>(noResizeCapacity);

        governor = new Semaphore(rateLimit);
	}

    public ConcurrentAgenda(final Collection<? extends AgendaItem> items) {
   		this(items, new AgendaConfig());
   	}

	private static void validateArguments(final Collection<?> items,
			final ThreadMonitor monitor, final int timeout) {
		if(items == null) {
			// empty is OK, we just won't do anything
			throw new IllegalArgumentException("Items must not be null");
		}
		if(monitor == null) {
			throw new IllegalArgumentException("Thread monitor must not be null");
		}
	}
	
	public synchronized int jobsLeft() {
		return todoItems.size() - completedItems;
	}

	public void run() {
		if(running()) {
			throw new IllegalStateException("Already running");
		}
		else {
			try { 
				init();
				queueAgendaItems();
				done.await();
			} catch (Exception e) {
				log.error("Error executing agenda items.", e);
				this.threadMonitor.setStatus(Status.FAILED);
			}
			finally {
				resetGovernor();
			}
		}
	}

	private void init() {
		completedItems = 0;
		cancelled = false;
		inProgressItems = new AtomicInteger();
		startedItems = new AtomicInteger();
		setTimeout();
	}

	public boolean isLaxErrors() {
		return laxErrors;
	}

	public void setLaxErrors(boolean lax) {
		laxErrors = lax;
	}

	/**
	 * Queues all agenda items that are in the <code>todoItems</code> list for
	 * concurrent executuion.
	 *
	 * This method uses the governor to ensure that at most [<code>RATE_LIMIT</code>]
	 * agenda items are queued for execution at any time.
	 * 
	 * @throws InterruptedException if the governor is interrupted while waiting
	 * for a permit in the semaphore to insert the current agenda item.
	 */
	private void queueAgendaItems() throws InterruptedException {
		
		final int numItems = todoItems.size();
		done = new CountDownLatch(numItems);
		if(numItems > 0) {
			this.threadMonitor.setStatus(Status.RUNNING);
			Iterator<AgendaItem> itemsToQueue = todoItems.iterator();

			while (itemsToQueue.hasNext() && !cancelled) {
				AgendaItem item = itemsToQueue.next();

				// make sure we don't have too many running already
				this.governor.acquire();

				// queue the item
				synchronized (this) {
					queuedItems.put(item,
						taskManager.execute(
							new FailsafeRunnable(item)));
				}			
			}

			log.debug("Submitted all items");
		}
		else {
			this.threadMonitor.setStatus(Status.FINISHED);
		}
	}

	private boolean running() {
		return done != null && done.getCount() > 0;
	}

	protected void fireListenerEvent(AgendaEvent event) {
		for (AgendaListener listener : listeners) {
			try {
				listener.handleEvent(event);
			}
			catch(Exception e) {
				// can't rethrow, have to notify other listeners
				log.error("Couldn't notifiy listener of event", e);
			}
		}
	}

	public void abort() {
		log.info("External caller aborted agenda");
		cancelTimeout();
		cancelPending(Status.ABORTED);
	}

	private void timeout() {
		log.info("Agenda timed out");
		cancelPending(Status.TIMEDOUT);
	}

	private void cancelPending(Status status) {
		if(running()) {
			cancelled = true;
			synchronized (this) {
				this.threadMonitor.setStatus(status);
				if(log.isDebugEnabled()) {
					log.debug("Cancelling " + queuedItems.size()
						+ " queued agenda items, " + inProgressItems.get()
						+ " of which are in progress");
				}
				Iterator<Map.Entry<AgendaItem, Future<?>>> entries =
						queuedItems.entrySet().iterator();
				while(entries.hasNext()) {
					Map.Entry<AgendaItem, Future<?>> entry =
						entries.next();

					AgendaItem item = entry.getKey();
					Future<?> future = entry.getValue();
					boolean didit = future.cancel(true);

					if(didit) {
						// NOTE: completeItem() will still be
						// called by the FailsafeRunnable for any
						// items which were actually started, so
						// it will decrement the latch (but see
						// finishCountingDown())
						entries.remove();

						if(log.isDebugEnabled()) {
							log.debug("Cancelled item " + item);
						}
					}
					else {
						if(log.isDebugEnabled()) {
							log.debug("Item " + item
								+ " already cancelled or completed");
						}
					}
				}
			}
			finishCountingDown();
		}
	}

	private void finishCountingDown() {
		// the downside of using a latch vs. the previous design of waiting for
		// a collection of futures is that we need to make sure that we always
		// decrement all the way -- here we count down for the threads that
		// were cancelled before they were ever started
		final int unstarted = todoItems.size() - startedItems.get();
		for(int i = 0; i < unstarted; i++) {
			done.countDown();
		}
	}

	private void resetGovernor() {
		if(governor.availablePermits() < rateLimit) {
			governor.release(rateLimit - governor.availablePermits());
		}
	}

	public ThreadMonitor getThreadMonitor() {
		return threadMonitor;
	}

	public synchronized void addListener(AgendaListener listener) {
		this.listeners.add(listener);
	}

	public synchronized void removeListener(AgendaListener listener) {
		this.listeners.remove(listener);
	}

	private void completeItem(final AgendaItem item, Exception e) {
		if(log.isDebugEnabled()) {
			log.debug("Completing item " + item);
		}
		try {
			inProgressItems.decrementAndGet();
			governor.release();
			EventType type = null;

			synchronized (this) {
				final Status status = threadMonitor.getStatus();
				++completedItems;
				item.setEndTime(System.currentTimeMillis());

				Future<?> future = queuedItems.remove(item);
				if(future == null) {
					// this item was already removed due to
					// being cancelled

					// FIXME: the following needs to be conditional
					// on the actual action not being allowed to fail,
					// but for that to occur, the Action/Activity/JobSet
					// API would need to be cleaned up (so we have some
					// way to know whether an individual acton can fail).
					// For now, it's all-or-nothing at the mashup level
					if(!laxErrors) {
						// NOTE: we infer the exception here for two reasons:
						// 1) we want timed-out tasks to have a TimeoutException,
						// NOT InterruptedException, and; 2) the task might
						// have actually swallowed the InterruptedException
						// (say, if it did any blocking thread operations
						// internally in its run() method)
						Exception inferred = null;
						if(Status.TIMEDOUT == status) {
							type = EventType.AgendaItemTimedout;
							inferred = new TimeoutException();
						}
						else {
							// we were either aborted or cancelled
							// due to the failure of another task
							type = EventType.AgendaActionItemCompleted;
							inferred = new InterruptedException();
						}
						setActionException(item, inferred);
					}
				}
				else {
					type = EventType.AgendaActionItemCompleted;
					if(e != null && !laxErrors) {
						// if we're not configured to ignore exceptions, make
						// sure the action reports the error
						setActionException(item, e);

						// and short circuit the rest of the tasks
						// NOTE: I don't think it should be possible to
						// get here w/ another status, but better safe
						// than sorry
						if(Status.RUNNING == status) {
							cancelTimeout();
							cancelPending(Status.FAILED);
						}
					}
					else if(completedItems == todoItems.size()) {
						cancelTimeout();
						this.threadMonitor.setStatus(
								Status.FINISHED);
					}
				}
				fireListenerEvent(new AgendaEvent(item, type));
			}

		}
		finally {
			// if we don't do this, main thread waits forever!
			done.countDown();
		}
	}

	private void setActionException(AgendaItem item, Exception e) {
		// FIXME: this whole mess shouldn't be necessary, except there
		// is no invariant forced by the Action/Activity API -- some
		// actions and activities swallow exceptions and set their output
		// appropriately, some throw exceptions, etc. The only case
		// where we should need to call this method is on timeout (which
		// the Action obviously knows nothing about)
		if(!laxErrors) {
			Runnable something = item.getItem();
			if(something instanceof Action) {
				Action action = (Action) something;
				ActionOutput output = action.getActionResult();
				if(output == null) {
					output = new ActivityResultItem(
							new ActionException(action.getName(),
								e.getMessage(), e));
				}
			}
			else {
				log.error("Don't know how to set error status for " +
						"task of type " + something.getClass().getName());
			}
		}
	}


	private void startItemExecution() {
		inProgressItems.incrementAndGet();
		startedItems.incrementAndGet();
	}

	private synchronized void setTimeout() {
		this.scheduledTimeout = this.taskManager.scheduleTimeout(new TimeoutTask(this), 
				this.agendaTimeout, TimeUnit.MILLISECONDS);
	}

	private synchronized void cancelTimeout() {
		if(this.scheduledTimeout != null) {
			this.scheduledTimeout.cancel(true);
		} else {
			log.warn("Trying to cancel a null scheduled timeout.");
		}
	}

	private class FailsafeRunnable implements Runnable {

		private final AgendaItem realItem;

		private FailsafeRunnable(AgendaItem runnable) {
			this.realItem = runnable;
		}

		public void run() {
			startItemExecution();
			try {
				realItem.run();
				completeItem(realItem, null);
			}
			catch(Exception e) {
				completeItem(realItem, e);
			}
		}
	}

	static class TimeoutTask implements Runnable {

		private ConcurrentAgenda agenda;

		public TimeoutTask(ConcurrentAgenda agenda) {
			this.agenda = agenda;
		}

		@Override
		public void run() {
			this.agenda.timeout();
		}

	}
}
